package zero.tech.redis.topic.config;

import java.util.ArrayList;
import java.util.List;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;

import com.fasterxml.jackson.annotation.JsonAutoDetect;
import com.fasterxml.jackson.annotation.PropertyAccessor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectMapper.DefaultTyping;

import zero.tech.core.util.EngineLogUtil;
import zero.tech.redis.topic.TopicCMD;
import zero.tech.redis.topic.TopicChannelType;
import zero.tech.redis.topic.TopicManager;
import zero.tech.redis.topic.service.api.IRedisMsgService;
@Configuration
@ConfigurationProperties(prefix = "redis.topic")
public class TopicConfigration {
	@Autowired
	private IRedisMsgService redisMsgService;
	private String sayhello="测试发送topic";
	private boolean enable=false;
	private Integer testSendInterval=120000;
	
	public Integer getTestSendInterval() {
		return testSendInterval;
	}

	public void setTestSendInterval(Integer testSendInterval) {
		this.testSendInterval = testSendInterval;
	}

	public String getSayhello() {
		return sayhello;
	}

	public void setSayhello(String sayhello) {
		this.sayhello = sayhello;
	}

	public boolean isEnable() {
		return enable;
	}

	public void setEnable(boolean enable) {
		this.enable = enable;
	}

	@SuppressWarnings({ "rawtypes", "unchecked", "deprecation" })
	@Bean
	@ConditionalOnProperty(prefix ="redis.topic",name="enable",havingValue = "true",matchIfMissing = false)
	public RedisMessageListenerContainer redisMessageListenerContainer(@Qualifier("topiclistener")RedisMessageListener listener,RedisConnectionFactory redisConnectionFactory) {
		RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
		redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
		redisMessageListenerContainer.addMessageListener(listener, topic());
		Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class);
	    ObjectMapper om = new ObjectMapper();
	    om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
	    om.enableDefaultTyping(DefaultTyping.NON_FINAL);
	    jackson2JsonRedisSerializer.setObjectMapper(om);
	    redisMessageListenerContainer.setTopicSerializer(jackson2JsonRedisSerializer);
		return redisMessageListenerContainer;
	}

	@Bean("topiclistener")
	@ConditionalOnProperty(prefix ="redis.topic",name="enable",havingValue = "true",matchIfMissing = false)
	public RedisMessageListener listener(RedisTemplate<String, Object> redisTemplate) {
		return new RedisMessageListener(redisTemplate,TopicManager.getManager());
	}

	public List<ChannelTopic> topic() {
		List<ChannelTopic> topics = new ArrayList<ChannelTopic>();
		TopicChannelType[] topicArray=TopicChannelType.values();
		for(TopicChannelType type:topicArray) {
			ChannelTopic topic = new ChannelTopic(
					type.getChannelType());
			topics.add(topic);
		}
		return topics;
	}
	
	
	protected class Init{
	}
	
	@Bean("inittopic")
	@ConditionalOnProperty(prefix ="redis.topic",name="enable",havingValue = "true",matchIfMissing = false)
	public Init init() {
		try {
			TopicManager.getManager().start();
		} catch (Exception e) {
			System.exit(1);
			e.printStackTrace();
		}
		initTopic();
		return new Init();
	}

	private void initTopic() {
		new Thread(new Runnable() {
			@Override
			public void run() {
				while (true) {
					try {
						redisMsgService.pulishRedisTopic(
								TopicChannelType.DEFAULT,
								TopicCMD.TEST,sayhello);
					} catch (Exception e) {
						EngineLogUtil.traceFull(e);
					}
					try {
						Thread.sleep(testSendInterval);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		}).start();
	}
}
